package com.ky.pro.java.mq.kmq.core;

import lombok.SneakyThrows;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class KmqConsumer<T> {

    private final KmqBroker broker;

    private KQueue<T> kmq;

    String consumerId;

    public KmqConsumer(KmqBroker broker) {
        this.broker = broker;
        // consumerId：简单实现，后续可以考虑引入分布式id、用户自定义全局ID等
        consumerId = UUID.randomUUID().toString();
    }

    public void subscribe(String topic) {
        this.kmq = this.broker.findKmq(topic);
        if (null == kmq) throw new RuntimeException("Topic[" + topic + "] doesn't exist.");
        kmq.addConsumer(consumerId);
    }

    public void unsubscribe(String topic) {
        this.kmq = this.broker.findKmq(topic);
        if (null == kmq) throw new RuntimeException("Topic[" + topic + "] doesn't exist.");
        kmq.removeConsumer(consumerId);
    }

    @SneakyThrows
    public List<KmqMessage<T>> poll(long timeout, TimeUnit timeUnit, int batchSize) {
        return kmq.poll(consumerId, timeout, timeUnit, batchSize);
    }

}
